package org.lzfto.flink.demo.business.flinkTest.flinkObjects.function.process;

import java.math.BigDecimal;

import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.lzfto.flink.demo.business.orders.OrderCommon;
import org.lzfto.flink.demo.business.orders.SimpleOrder;
import org.lzfto.flink.demo.business.orders.SimpleOrderAgg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
*订单分组处理函数
*@author lzfto
*@创建时间 2025年6月20日 09:39:28
*@param T  type of key
*/
public class  OrderKeyedProcessFunction<T> extends KeyedProcessFunction<T, SimpleOrder, SimpleOrderAgg> {
	
	private final Logger LOGGER = LoggerFactory.getLogger(getClass());
	
	private static final long serialVersionUID = 1L;
	private String grpType;
	private transient ValueState<SimpleOrderAgg> agg = null;
	private transient ValueState<Long> timerState;
	// 30秒无输入则清除agg,避免浪费内存,但是这样就等同于只有一个小时的窗口
	private static final long INACTIVITY_TIMEOUT = 30 * 1000L; 

	
	public OrderKeyedProcessFunction(String grpType) {
		super();
		this.grpType=grpType;
	}
	
	
	@Override
	public void open(OpenContext openContext) {
		// 初始化统计值
		TypeInformation<SimpleOrderAgg> aggType = TypeInformation.of(SimpleOrderAgg.class);
		ValueStateDescriptor<SimpleOrderAgg> aggDescriptor = new ValueStateDescriptor<>("agg", aggType);
		agg = getRuntimeContext().getState(aggDescriptor);
		
		ValueStateDescriptor<Long> timeStateDescriptor = new ValueStateDescriptor<>("timerState", Long.class);
		timerState=getRuntimeContext().getState(timeStateDescriptor);
	}

	@Override
	public void processElement(SimpleOrder value, Context ctx, Collector<SimpleOrderAgg> out)
			throws Exception {
		SimpleOrderAgg tmp = agg.value();
		if (tmp == null) {
	        tmp = new SimpleOrderAgg();	        
	        tmp.setCount(0);
	        tmp.setTotal(BigDecimal.ZERO);
	        if (grpType.equals(OrderCommon.GRP_TYPE_SEX)) {
	        	tmp.setSex(value.getSex());	
	        }
	        else if (grpType.equals(OrderCommon.GRP_TYPE_ZONE)) {
	        	tmp.setZoneName(value.getZoneName());	
	        }
	        else {
	        	tmp.setZoneName(value.getZoneName());
		        tmp.setSex(value.getSex());	
	        }
	    }
		tmp.setCount(tmp.getCount() + 1);
		tmp.setTotal(tmp.getTotal().add(value.getPrice()));
		agg.update(tmp);
		out.collect(agg.value());
		
		// 注册/更新超时清理定时器
        long currentTime = ctx.timerService().currentProcessingTime();
        long timeoutTimestamp = currentTime + INACTIVITY_TIMEOUT;

        Long oldTimer = timerState.value();
        if (oldTimer != null) {
            ctx.timerService().deleteProcessingTimeTimer(oldTimer);
        }
        ctx.timerService().registerProcessingTimeTimer(timeoutTimestamp);
        timerState.update(timeoutTimestamp);
	}
	
	@Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<SimpleOrderAgg> out) throws Exception {
        // 当前定时器已过期，清除状态
        agg.clear();
        timerState.clear();
        LOGGER.warn("状态已清除: key = " + ctx.getCurrentKey());
    }
}
